package com.zhao.biz.report.report2_region.version1_sql.way1_single

import com.typesafe.config.Config
import com.zhao.entity.Log
import com.zhao.utils.CommonUtil
import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

import java.util.Properties

/**
 * Description: 报表展示之地域分布spark sql版本<br/>
 * Copyright (c) ，2021 ， 赵 <br/>
 * A wet person does not fear the rain. <br/>
 * Date： 2021/1/13 14:08
 *
 * @author 柒柒
 * @version : 1.0
 */

object RegionCal {
  def main(args: Array[String]): Unit = {

    val inputPath = "a_data/outputpath"
    val Array(input) = Array(inputPath)

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      //设置序列化技术(使用kryo)
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext

    //2.注册实体类
    sc.getConf.registerKryoClasses(Array(classOf[Log]))

    //3.加载ETL后的数据到内存中,装载为DataFrame,并映射为一张虚拟表
    spark.read.parquet(inputPath).createOrReplaceTempView("tb_log")

    //缓存
    spark.sqlContext.cacheTable("tb_log")

    //4.使用sparksql统计地域分布
    //requestmode  数据请求方式（1:请求、2:展示、3:点击）
    //procenode 流程节点（1：请求量 kpi 2：有效请求 3：广告请求）
    //iseffective  有效标识（有效指可以正常计费的）(0：无效 1：有效)
    //isbilling  是否收费（0：未收费 1：已收费）
    //isbid  是否 rtb
    //iswin  是否竞价成功
    //adorderid  广告id
    //adpayment  转换后广告消费
    //winprice   rtb竞争成功价格
    val resultDF: DataFrame = spark.sql(
      """
        |select
        | provincename `省份名`,
        | cityname `城市名`,
        | originalReq,
        | effectiveReq,
        | adReq,
        | bidCnt,
        | bidSuccCnt,
        | concat(cast(if(bidSuccCnt=0,0,bidSuccCnt*100.0/bidCnt) as decimal(10,2)),'%') bidSuccRate,
        | showCnt,
        | clickCnt,
        | if(clickCnt=0,0,clickCnt*1.0/showCnt) clickRate,
        | adCost,
        | adConsumer
        |from(
        |    select
        |    provincename,
        |    cityname,
        |    sum(case when requestmode = 1 and processnode >= 1 then 1 else 0 end) originalReq, -- 原始请求数据
        |    sum(case when requestmode = 1 and processnode >= 2 then 1 else 0 end) effectiveReq, -- 有效请求数据
        |    sum(case when requestmode = 1 and processnode = 3 then 1 else 0 end) adReq, -- 广告请求数据
        |    sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 then 1 else 0 end) bidCnt, -- 参与竞价数
        |    sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 and adorderid != 0 then 1 else 0 end) bidSuccCnt, -- 竞价成功数
        |    sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end) showCnt, -- 展示量
        |    sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end) clickCnt, -- 点击量
        |    sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then adpayment/1000 else 0 end) adCost, --广告成本(广告公司支付给网站主(媒体))
        |    sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then winprice/1000 else 0 end) adConsumer -- 广告消费(广告主给广告公司)
        |  from tb_log
        |group by provincename,cityname) t
        |""".stripMargin)

    //5.落地
    //测试:显示到控制台
    resultDF.show(10000)

    //落地到db中
    //保存到mysql中
//    val load: Config = CommonUtil.load
//    //设置请求mysql的配置信息
//    val prop: Properties = new Properties()
//    prop.setProperty("user",load.getString("db.default.user"))
//    prop.setProperty("password",load.getString("db.default.password"))
//    resultDF.write.mode(SaveMode.Overwrite)
//      .jdbc(load.getString("db.default.url"),load.getString("db.default.tbname2"),prop)

    //资源释放
    spark.stop()
  }
}

















